// 回显服务把收到的数据包发送给客户端。 增加了一个流量统计功能和其他系统信息。
#include <muduo/base/Logging.h>
#include <muduo/base/Atomic.h>
#include <muduo/base/FileUtil.h>
#include <muduo/base/ProcessInfo.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/InetAddress.h>
#include <boost/bind.hpp>
#include <stdio.h>

using namespace muduo;
using namespace muduo::net;

static int numThreads = 0;

// RFC 862
class EchoServer {
public:
	EchoServer(EventLoop *loop, const InetAddress &listenAddr) 
		: server_(loop, listenAddr, "EchoServer"),
		  oldCounter_(0),
		  startTime_(Timestamp::now()) {
		server_.setConnectionCallback(boost::bind(&EchoServer::OnConnection, this, _1));
		server_.setMessageCallback(boost::bind(&EchoServer::OnMessage, this, _1, _2, _3));
		server_.setThreadNum(numThreads);
		loop->runEvery(5.0, boost::bind(&EchoServer::PrintThrought, this));
	}

	void Start() {
		LOG_INFO << "*** start with " << numThreads << " threads";
		server_.start();
	}

private:
	void OnConnection(const TcpConnectionPtr &conn) {
		LOG_INFO << conn->peerAddress().toIpPort() << " -> "
		        << conn->localAddress().toIpPort() << " is "
		        << (conn->connected() ? "UP" : "DOWN");
		conn->setTcpNoDelay(true);
		if (conn->connected()) {
      		connections_.increment();
      	} else {
      		connections_.decrement();
      	}
	}

	void OnMessage(const TcpConnectionPtr &conn, Buffer *buf, Timestamp time) {
		size_t len = buf->readableBytes();
		transferred_.add(len);
		receivedMessages_.incrementAndGet();
		conn->send(buf);
	}

	void PrintThrought() {
		Timestamp endTime = Timestamp::now();
		int64_t newCounter = transferred_.get();
		int64_t bytes = newCounter - oldCounter_;
		int64_t msgs = receivedMessages_.getAndSet(0);
		double time = timeDifference(endTime, startTime_);
		printf("*** %4.3f MiB/s %4.3f Ki Msgs/s %6.2f bytes per msg\n",
        	static_cast<double>(bytes)/time/1024/1024,
        	static_cast<double>(msgs)/time/1024,
        	static_cast<double>(bytes)/static_cast<double>(msgs));

		PrintConnection();
		fflush(stdout);
		oldCounter_ = newCounter;
		startTime_ = endTime;
	}

	void PrintConnection() {
		string procStatus = ProcessInfo::procStatus();
		printf("%d conn, files %d , VmSize %ld KiB, RSS %ld KiB, ",
		       connections_.get(),
		       ProcessInfo::openedFiles(),
		       GetLong(procStatus, "VmSize:"),
		       GetLong(procStatus, "VmRSS:"));

		string meminfo;
		FileUtil::readFile("/proc/meminfo", 65536, &meminfo);
		long total_kb = GetLong(meminfo, "MemTotal:");
		long free_kb = GetLong(meminfo, "MemFree:");
		long buffers_kb = GetLong(meminfo, "Buffers:");
		long cached_kb = GetLong(meminfo, "Cached:");
		printf("system memory used %ld KiB\n",
		       total_kb - free_kb - buffers_kb - cached_kb);
	}

	long GetLong(const string& procStatus, const char* key) {
		long result = 0;
		size_t pos = procStatus.find(key);
		if (pos != string::npos)
		{
		  result = ::atol(procStatus.c_str() + pos + strlen(key));
		}
		return result;
	}

	TcpServer server_;
	AtomicInt32 connections_;
	AtomicInt64 transferred_;
	AtomicInt64 receivedMessages_;
	int64_t oldCounter_;
	Timestamp startTime_;
};

int 
main(int argc, char **argv) {
	Logger::setLogLevel(Logger::INFO);
	LOG_INFO << "pid = " << getpid() << ", tid = " << CurrentThread::tid();
	if (argc > 1)
		numThreads = atoi(argv[1]);
	EventLoop loop;
	InetAddress listenAddr(2009);
	EchoServer server(&loop, listenAddr);
	server.Start();
	loop.loop();
}
